package com.chb.flink.func

import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet}
import java.text.SimpleDateFormat

import com.chb.flink.source.StationLog
import org.apache.flink.api.common.functions.{MapFunction, RichMapFunction}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

/**
 * 富函数类举例：把呼叫成功的通话信息转化成真实的用户姓名
 */
object TestRichFunction {
    def main(args: Array[String]): Unit = {
        val streamEnv = StreamExecutionEnvironment.getExecutionEnvironment

        // 隐式转换
        import org.apache.flink.streaming.api.scala._

        val data: DataStream[StationLog] = streamEnv.readTextFile(getClass.getResource("/station.log").getPath)
            .map(line => {
                val arr = line.split(",")
                new StationLog(arr(0).trim, arr(1).trim, arr(2).trim, arr(3).trim, arr(4).trim.toLong, arr(5).trim.toLong)
            })

        //定义时间输出格式
        val format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")

        //过滤那些通话成功的
        data.filter(_.callType.equals("success"))
            .map(new CallRichMapFunction)
            .print()

        streamEnv.execute()
    }
}


//自定义的富函数类
class CallRichMapFunction() extends RichMapFunction[StationLog, StationLog] {
    var conn: Connection = _
    var pst: PreparedStatement = _

    //生命周期管理， 初始化时，创建数据库连接
    override def open(parameters: Configuration): Unit = {
        conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/test?useSSL=false", "root", "123456")
        pst = conn.prepareStatement("select name from t_phone where phone_number = ?")
    }

    override def map(in: StationLog): StationLog = {
        pst.setString(1, in.callOut)
        var set1 = pst.executeQuery()
        if (set1.next()) {
            in.callOut = set1.getString(1)
        }

        // 查找被叫名字
        pst.setString(1, in.callIn)
        val set2 = pst.executeQuery()
        if (set2.next()) {
            in.callIn = set2.getString(1)
        }
        in
    }

    override def close(): Unit = {
        pst.close()
        conn.close()
    }

}
